package com.flink.timewindow.watermark;

import com.flink.timewindow.partition.CustomPartitioner;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
//设置空闲等待
public class WatermarkIdlenessDemo {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         //并行度为2
        env.setParallelism(2);
        //获取数据源
        //自定义分区器：数据%分区数 ，只输入奇数，都只会取往map的一个子任务
        SingleOutputStreamOperator<Integer> socketDS = env
                .socketTextStream("10.90.100.102", 8888)
                .partitionCustom(new CustomPartitioner(),key->key)
                //转换
                .map(Integer::valueOf)
                .assignTimestampsAndWatermarks(
                                WatermarkStrategy
                                        .<Integer>forMonotonousTimestamps()
                                        .withTimestampAssigner((v,ts)->v*1000L)
                                        //空闲等待5s
                                        .withIdleness(Duration.ofSeconds(5))
                        );



        //分成两组：奇数一组，偶数一组，开10s的事件时间滚动窗口
        socketDS
                .keyBy(key->key%2)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(new ProcessWindowFunction<Integer, String, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer integer, ProcessWindowFunction<Integer, String, Integer, TimeWindow>.Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
                        //上下文可以拿到window对象，还有其他东西：如测流输出
                        long count = elements.spliterator().estimateSize();
                        long windowStartTs = context.window().getStart();
                        long windowEndTs = context.window().getEnd();
                        String windowStart = DateFormatUtils.format(windowStartTs, "yyyy-MM-dd HH:mm:ss.SSS");
                        String windowEnd = DateFormatUtils.format(windowEndTs, "yyyy-MM-dd HH:mm:ss.SSS");

                        out.collect("key=" + integer + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());

                    }
                })
                .print();
        //执行
        env.execute();

    }
}
